package org.apereo.cas.web.y9.event.listener.kafka;

import java.util.List;
import java.util.Set;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apereo.cas.web.y9.event.message.Y9MessageCommon;
import org.apereo.cas.web.y9.event.message.Y9MessageOrg;
import org.apereo.cas.web.y9.event.message.Y9OrgConst;
import org.apereo.cas.web.y9.util.Y9Context;
import org.apereo.cas.web.y9.util.json.Y9JacksonUtil;
import org.apereo.cas.web.y9.y9user.Y9User;
import org.apereo.cas.web.y9.y9user.Y9UserDao;
import org.springframework.cloud.context.refresh.ContextRefresher;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class Y9OrgEventListener {

    private ContextRefresher contextRefresher;
    private Y9UserDao y9UserDao;

    @KafkaListener(topics = {"y9_common_event"})
    public void messageCommonListener4kafka(ConsumerRecord<String, String> data) {
        String value = data.value().toString();
        Y9MessageCommon msg = null;
        try {
            msg = Y9JacksonUtil.readValue(value, Y9MessageCommon.class);
            String eventType = msg.getEventType();
            if (Y9MessageCommon.REFRESH_REMOTE_APPLICATION_EVENT.equals(eventType)) {
                if (contextRefresher == null) {
                    try {
                        this.contextRefresher = Y9Context.getBean(ContextRefresher.class);
                    } catch (Exception e) {
                        LOGGER.error("contextRefresher==null，刷新bean信息失败。", e);
                    }
                }

                if (contextRefresher != null) {
                    Set<String> keys = contextRefresher.refresh();
                    LOGGER.info("Received remote refresh request. Keys refreshed " + keys);
                }
            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    @KafkaListener(topics = {"y9_user_event"})
    public void messageOrgListener4kafka(ConsumerRecord<String, String> data) {
        String value = data.value().toString();
        Y9MessageOrg msg = null;
        try {
            if (y9UserDao == null) {
                y9UserDao = Y9Context.getBean(Y9UserDao.class);
            }
            msg = Y9JacksonUtil.readValue(value, Y9MessageOrg.class);
            if (Y9OrgConst.RISEORGEVENT_TYPE_UPDATE_USER.equals(msg.getEventType())) {
                LOGGER.info("--------------------------添加或修改人员-------------------------------");
                Y9User y9User = Y9JacksonUtil.readValue(msg.getOrgObj().toString(), Y9User.class);
                // y9UserDao.save(y9User);
                LOGGER.info("--------------------" + y9User.getName() + "---------------------");
            } else if (Y9OrgConst.RISEORGEVENT_TYPE_DELETE_USER.equals(msg.getEventType())) {
                LOGGER.info("--------------------------删除人员-------------------------------");
                Y9User y9User = Y9JacksonUtil.readValue(msg.getOrgObj().toString(), Y9User.class);
                LOGGER.info("--------------------" + y9User.getName() + "---------------------");
                // y9UserDao.delete(y9User);
            } else if (Y9OrgConst.RISEORGEVENT_TYPE_USER_SYNC.equals(msg.getEventType())) {
                LOGGER.info("--------------------------同步人员-------------------------------");
                // y9UserDao.deleteAllInBatch();
                String json = msg.getOrgObj().toString();
                List<Y9User> users = Y9JacksonUtil.readList(json, Y9User.class);
                for (Y9User y9User : users) {
                    LOGGER.info("--------------------" + y9User.getName() + "---------------------");
                    // y9UserDao.save(y9User);
                }
            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

}
